package njit.mykafka;

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.*;

public class Combine implements Runnable {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public Combine(String name) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("retries", 0);
        props.put("acks", "all");
        props.put("batch.size", 16384);
        props.put("buffer.memory", 333554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = name;
    }

    public void run() {
        int messageNo = 1;
        try {
            for (;;) {
                String messageStr = "你好，这是第" + messageNo + "条数据";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                if (messageNo % 100 == 0)
                    System.out.println("发送的消息:" + messageStr);
                if (messageNo % 1000 == 0) {
                    System.out.println("成功发送了" + messageNo + "条");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String[] args) {
        Combine combine = new Combine("my_test");
        Thread thread = new Thread(combine);
        thread.start();
        KafkaConsumerTest test = new KafkaConsumerTest("my_test");
        Thread thread2 = new Thread(test);
        thread2.start();
    }
}

class KafkaConsumerTest implements Runnable {
    private final KafkaConsumer<String, String> consumer;
    private final String topic;
    private ConsumerRecords<String, String> msgList;

    public KafkaConsumerTest(String name) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("group.id", "groupA");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = name;
        this.consumer.subscribe(Arrays.asList(topic));
    }

    public void run() {
        int messageNo = 1;
        System.out.println("-------开始消费--------");
        try {
            for (;;) {
                msgList = consumer.poll(1000);
                if (null != msgList && msgList.count() > 0)
                    for (ConsumerRecord<String, String> record : msgList) {
                        if (messageNo % 100 == 0)
                            System.out.println(messageNo + "===================receive:key= " + record.key()
                                    + ",value= " + record.value() + " offset===" + record.offset());
                        if (messageNo % 1000 == 0)
                            break;
                        messageNo++;
                    }
                else
                    Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}
